package com.orderstats.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.sql.*;
import java.time.Duration;
import java.util.*;

public class DirectKafkaToDatabase {
    
    private static final String TOPIC = "orders";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String GROUP_ID = "direct-consumer";
    private static final String DB_PATH = "/home/ubuntu/orders_system.db";
    
    public static void main(String[] args) {
        System.out.println("🚀 启动直接Kafka到数据库消费者...");
        
        // 初始化数据库
        initDatabase();
        
        // 配置Kafka消费者
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList(TOPIC));
            System.out.println("📊 开始实时消费订单数据...");
            
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        String orderData = record.value();
                        System.out.println("📥 接收订单: " + orderData);
                        
                        // 解析并保存订单数据
                        if (parseAndSaveOrder(orderData)) {
                            System.out.println("✅ 订单已保存到数据库");
                        }
                    } catch (Exception e) {
                        System.err.println("❌ 处理订单失败: " + e.getMessage());
                    }
                }
                
                // 每10秒显示一次统计
                if (System.currentTimeMillis() % 10000 < 1000) {
                    showStats();
                }
            }
        } catch (Exception e) {
            System.err.println("❌ 消费者异常: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    private static void initDatabase() {
        String createTableSQL = "CREATE TABLE IF NOT EXISTS orders (" +
            "id INTEGER PRIMARY KEY AUTOINCREMENT, " +
            "category TEXT NOT NULL, " +
            "product TEXT NOT NULL, " +
            "user_id TEXT NOT NULL, " +
            "quantity INTEGER NOT NULL, " +
            "order_time TEXT NOT NULL, " +
            "rating REAL NOT NULL, " +
            "is_valid TEXT NOT NULL, " +
            "created_at DATETIME DEFAULT CURRENT_TIMESTAMP" +
            ")";
        
        try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + DB_PATH);
             Statement stmt = conn.createStatement()) {
            stmt.execute(createTableSQL);
            System.out.println("✅ 数据库表已初始化");
        } catch (SQLException e) {
            System.err.println("❌ 数据库初始化失败: " + e.getMessage());
        }
    }
    
    private static boolean parseAndSaveOrder(String orderData) {
        try {
            // 解析CSV格式: category,product,user,quantity,orderTime,rating,isValid
            String[] fields = orderData.split(",");
            if (fields.length != 7) {
                System.err.println("❌ 数据格式错误: " + orderData);
                return false;
            }
            
            String category = fields[0].trim();
            String product = fields[1].trim();
            String userId = fields[2].trim();
            int quantity = Integer.parseInt(fields[3].trim());
            String orderTime = fields[4].trim();
            double rating = Double.parseDouble(fields[5].trim());
            String isValid = fields[6].trim();
            
            return saveToDatabase(category, product, userId, quantity, orderTime, rating, isValid);
            
        } catch (Exception e) {
            System.err.println("❌ 解析订单数据失败: " + e.getMessage());
            return false;
        }
    }
    
    private static boolean saveToDatabase(String category, String product, String userId, 
                                        int quantity, String orderTime, double rating, String isValid) {
        String sql = "INSERT INTO orders (category, product, user_id, quantity, order_time, rating, is_valid) VALUES (?, ?, ?, ?, ?, ?, ?)";
        
        try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + DB_PATH);
             PreparedStatement pstmt = conn.prepareStatement(sql)) {
            
            pstmt.setString(1, category);
            pstmt.setString(2, product);
            pstmt.setString(3, userId);
            pstmt.setInt(4, quantity);
            pstmt.setString(5, orderTime);
            pstmt.setDouble(6, rating);
            pstmt.setString(7, isValid);
            
            int result = pstmt.executeUpdate();
            return result > 0;
            
        } catch (SQLException e) {
            System.err.println("❌ 保存到数据库失败: " + e.getMessage());
            return false;
        }
    }
    
    private static void showStats() {
        try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + DB_PATH);
             Statement stmt = conn.createStatement()) {
            
            ResultSet rs = stmt.executeQuery("SELECT COUNT(*) as total, COUNT(CASE WHEN is_valid='Y' THEN 1 END) as valid FROM orders");
            if (rs.next()) {
                int total = rs.getInt("total");
                int valid = rs.getInt("valid");
                System.out.println("📈 数据库统计: 总订单=" + total + ", 有效订单=" + valid);
            }
        } catch (SQLException e) {
            System.err.println("❌ 查询统计失败: " + e.getMessage());
        }
    }
} 